package com.atguigu.edu.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.Base.BaseStreamApp;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.utils.FlinkSinkUtil;
import com.atguigu.edu.realtime.utils.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/** 维度表
 分类表（base_category_info）
 知识点问题表（test_point_question）
 省份表（base_province）
 来源表（base_source）
 科目表（base_subject_info）
 章节表（chapter_info）
 课程信息表（course_info）
 知识点表（knowledge_point）
 用户表（user_info）
 视频表（video_info）



 */

@Slf4j
public class DimApp extends BaseStreamApp {

    public static void main(String[] args) {
        new DimApp().init(
                2221,
                2,
                "DimApp",
                Constant.TOPIC_ODS_DB);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 实现业务: 对流做各种操作
        // 1. 清洗数据
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);

        // 2. 读取配置信息
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);

//        // 3. 根据配置信息, 进行建表或删表
        tpStream = createOrDropTable(tpStream);

        // 4. 过滤出需要的数据, 分流(动态分流技术 cdc connect 广播状态 )
        // 给每条维度数据搭配一个配置信息
        // sku_info 的维度数据, 写入到维度表中:
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimToConfigStream = connect(etledStream, tpStream);

        // 5. 删除不需要的列
        dimToConfigStream = delNotNeedColumns(dimToConfigStream);
        dimToConfigStream.print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 6. 写出到到 Phoenix 中
        // bin/maxwell-bootstrap --config config.properties --database edu --table base_trademark
        // 启动phoenix:  /opt/module/phoenix/bin/sqlline.py hadoop162,hadoop163,hadoop164:2181。命令!table查看所有表格
        writeToPhoenix(dimToConfigStream);
    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        /*
        如何写出到 phoenix 中?
        1. 找到 phoenix 连接器. 没有
        2. 使用 jdbc 连接器?
            不能. 因为 jdbc sink 只能把流中的数据写入到一个表中

        3. 只能 自定义 sink
         */
        stream.addSink(FlinkSinkUtil.getPhoenixSink());

    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNotNeedColumns(
            SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        return stream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                //                System.out.println(t);
                JSONObject data = t.f0;  // 本质就是 map 集合, 里面的一些 key 没有存在必要,要删除
                log.debug("JSONObject data = t.f0; >>>>>>>>>>>>>>>>"+data.toJSONString());
                List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));
                data.keySet().removeIf(key -> !columns.contains(key) && !"op_type".equals(key));

                return t;
            }
        });
    }

    // 连接数据流和配置流
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> dataStream,
            SingleOutputStreamOperator<TableProcess> tpStream) {
        // 1. 把配置流做成广播流
        // key: String   source_table:source_type
        // value: TableProcess
        MapStateDescriptor<String, TableProcess> tpStateDesc =
                new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 2. 让数据流去 connect 广播流
        return dataStream
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                    // 处理数据流中的数据: 维度数据
                    @Override
                    public void processElement(JSONObject obj,
                                               ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 4. 处理数据流中的数据, 从广播状态读取数据
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key = obj.getString("table") + ":ALL";

                        TableProcess tp = state.get(key);
                        // tp有没有可能是 null? 比如这条信息是事实表,或者是一个不需要的维度表
                        if (tp != null) {
                            JSONObject data = obj.getJSONObject("data");
                            data.put("op_type", obj.getString("type")); // 这个字段后面要用
                            out.collect(Tuple2.of(data, tp));
                        }
                    }

                    // 处理广播流中的数据: 配置信息
                    @Override
                    public void processBroadcastElement(TableProcess tp,
                                                        Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {

                        // 3. 把配置信息放入到广播状态中
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key = tp.getSourceTable() + ":" + tp.getSourceType();
                        // 不同的配置信息的操作, 对状态的操作应该不同
                        if ("d".equals(tp.getOp())) { // 配置信息删除了一条配置
                            state.remove(key);
                        } else {
                            state.put(key, tp);
                        }

                    }
                });


    }

    private SingleOutputStreamOperator<TableProcess> createOrDropTable(SingleOutputStreamOperator<TableProcess> tpStream) {
        return tpStream
                .filter(tp -> "dim".equals(tp.getSinkType()))  // 只过滤出维度表的配置信息
                .map(new RichMapFunction<TableProcess, TableProcess>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 1. 建立到 phoenix 的 jdbc 连接  ctrl+alt+f 快速变成一个成员变量
                        conn = JdbcUtil.getPhoenixConnection();

                    }

                    @Override
                    public void close() throws Exception {
                        // 2. 关闭连接
                        JdbcUtil.closeConnection(conn);
                    }

                    @Override
                    public TableProcess map(TableProcess tp) throws Exception {
                        System.out.println(tp);
                        String op = tp.getOp();
                        if ("r".equals(op) || "c".equals(op)) {
                            // 执行建表
                            createTable(tp);
                        } else if ("d".equals(op)) {
                            // 执行删表
                            dropTable(tp);
                        } else {
                            // 再执行删表
                            dropTable(tp);
                            // 先执行建表
                            createTable(tp);
                        }
                        return tp;
                    }

                    // 在 phoenix 中删表
                    private void dropTable(TableProcess tp) throws SQLException {
                        // 1. 拼接删表语句
                        String sql = "drop table if exists " + tp.getSinkTable();
                        System.out.println("删表语句: " + sql);
                        execSQL(sql);

                    }

                    // 在 phoenix 中建表
                    private void createTable(TableProcess tp) throws SQLException {
                        // 1. 拼接建表语句
                        // create table if not exists person(id varchar , name varchar, constraint primary key pk(id))SALT_BUCKETS = 4
                        // create table if not exists person(id varchar , name varchar, constraint primary key pk(id))
                        StringBuilder sql = new StringBuilder();
                        sql
                                .append("create table if not exists ")
                                .append(tp.getSinkTable())
                                .append("(")
                                // 拼接字段
                                .append(tp.getSinkColumns().replaceAll("[^,]+", "$0 varchar"))
                                // pk: 联合逐渐的名字
                                .append(", constraint pk primary key(")
                                // 拼接主键
                                .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                                .append("))")
                                .append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                        System.out.println("建表语句: " + sql);
                        execSQL(sql.toString());
                    }

                    private void execSQL(String sql) throws SQLException {
                        if (conn.isClosed()) {
                            // 如果连接由于长链接问题, 被自动关闭, 则获取一个新的链接
                            conn = JdbcUtil.getPhoenixConnection();
                        }
                        // 执行指定的 sql语句
                        // 2. 根据 conn 获取一个预处理语句
                        PreparedStatement ps = conn.prepareStatement(sql);
                        // 3. 执行 sql 语句
                        ps.execute();
                        // 4. 关闭预处理语句
                        ps.close();
                    }
                });


    }


    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        // 1. value是 json 格式
                        // 2. 数据库必须满足
                        try {
                            JSONObject obj = JSON.parseObject(value);

                            String type = obj.getString("type");

                            String data = obj.getString("data");

                            return "edu".equals(obj.getString("database"))
                                    && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))
                                    && data != null
                                    && data.length() > 2;
                        } catch (Exception e) {
                            log.warn("数据格式有误,不是正确的 json 数据: " + value);
                            return false;
                        }
                    }
                })
                .map(json -> JSON.parseObject(json.replaceAll("bootstrap-", "")));
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        // todo by me 坑点：不加 ("useSSL", "false") 会报这个错。什么原因不知道
        // todo The last packet sent successfully to the server was 0 milliseconds ago.The driver has not received any packets from the server.
        Properties props = new Properties();
        props.setProperty("useSSL", "false");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .jdbcProperties(props)
                .databaseList("edu_config") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("edu_config.table_process") // set captured table
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        /* by me
            读取的配置数据中 op:
            r: 当程序启动的时候, 读取的快照   before: null after: 有\
                建表
            u: 更新某个字段的时候  before:有  after: 有
                先删表,再建表
            c: 新增数据 before: null after: 有
                建表
            d: 删除数据 before: 有 after: null
                删表

            如果修改的是主键: 则先 d 再 c
        * */
        return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String json) throws Exception {
                        JSONObject obj = JSON.parseObject(json);
                        String op = obj.getString("op");
                        TableProcess tp;
                        if (!"d".equals(op)) {
                            tp = obj.getObject("after", TableProcess.class);
                        } else {
                            tp = obj.getObject("before", TableProcess.class);
                        }
                        // todo by me别忘了把操作类型加进去啊！！后面要用到，如果是d删除就要对应得把dim表删掉
                        tp.setOp(op);
                        return tp;
                    }
                });


    }
}


